package com.lx.sortsum;

import com.lx.entitys.CovidSumEntity;
import com.lx.sum.CovidSumMapper;
import com.lx.sum.CovidSumReducer;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: chenjiang
 * @Date: 2021/11/18/17:00
 * @Description: 运行提交程序入口
 */
@Slf4j
public class CovidSortSumDriver extends Configured implements Tool {
    public static void main(String[] args) {
        Configuration configuration = new Configuration();
        try {
            Path fileOutPath = new Path(args[1]);
            FileSystem fileSystem = FileSystem.get(configuration);
            if (fileSystem.exists(fileOutPath)) {
                fileSystem.delete(fileOutPath, true);
            }
            final int status = ToolRunner.run(configuration, new CovidSortSumDriver(), args);
            System.exit(status);
        } catch (Exception e) {
            log.error("程序运行错误:{}", e);
        }
    }

    @Override
    public int run(String[] args) throws Exception {

        final Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, CovidSortSumDriver.class.getSimpleName());

        job.setJarByClass(CovidSortSumDriver.class);

        job.setMapperClass(CovidSortSumMapper.class);
        job.setReducerClass(CovidSortSumReducer.class);

        job.setMapOutputKeyClass(CovidSumEntity.class);
        job.setMapOutputValueClass(Text.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(CovidSumEntity.class);

        Path input = new Path(args[0]);
        final Path output = new Path(args[1]);
        FileInputFormat.addInputPath(job, input);
        FileOutputFormat.setOutputPath(job, output);

        boolean result = job.waitForCompletion(true);

        return result ? 0 : 1;
    }
}
